package com.heima.comment.stream;

import com.alibaba.fastjson.JSON;
import com.heima.common.constants.ApCommentConstants;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.Aggregator;
import org.apache.kafka.streams.kstream.Initializer;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.TimeWindows;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;

import java.time.Duration;
import java.util.HashMap;
import java.util.Map;

/**
 * @author: Zqr
 * <p>
 * 实时计算评论的数据,(接收kafka消息topic= ApCommentConstants.AP_COMMENT_SYNC_ARTICLE ;
 * 流式计算后将组装好的数据再发送到topic=ApCommentConstants.AP_COMMENT_STREAM_OUT_SYNC_ARTICLE
 */
@Slf4j
@Component
public class ComputeCommentOnTime {


    //接收的数据格式是map格式的json字符串{"articleId":"1234","add":"xxx"}

    @Bean
    public KStream<String, String> kStream(StreamsBuilder streamsBuilder) {
        //从kafka的topic中接收消息用于流式计算
        KStream<String, String> stream = streamsBuilder.stream(ApCommentConstants.AP_COMMENT_SYNC_ARTICLE);

        stream.map((key, value) -> {
//                    System.out.println("初始value: " + value);
                    Map mp = JSON.parseObject(value, Map.class);
                    return new KeyValue<>(mp.get("articleId").toString(), ("add:" + mp.get("add")));
                })
                //根据文章id分组
                .groupBy((key, value) -> key)
                //时间窗口,每10秒计算一次
                .windowedBy(TimeWindows.of(Duration.ofSeconds(10)))
                //自定义的聚合计算
                .aggregate(new Initializer<String>() {
                    /**
                     * 初始方法,每个时间窗口内进行聚合前执行一次,也只会执行一次
                     * @return
                     */
                    @Override
                    public String apply() {
                        return "add:0";
                    }
                }, new Aggregator<String, String, String>() {
                    @Override
                    public String apply(String key, String value, String applyValue) {
                        System.out.println("key = " + key);
                        System.out.println("value = " + value);
                        System.out.println("applyValue = " + applyValue);
                        if (StringUtils.isBlank(value)) {
                            return applyValue;
                        }
                        //获取初始值
                        int commentCount = 0;
                        String[] split = applyValue.split(":");
                        commentCount = Integer.parseInt(split[1]);

                        /**
                         * 累加操作
                         */
                        System.out.println(">>>>>>>>value:= " + value);
                        String[] valAry = value.split(":");
                        commentCount += Integer.parseInt(valAry[1]);
                        String formatStr = String.format("add:%d", commentCount);
                        log.info("流式计算窗口内文章评论数,当前文章id:{}", key);
                        log.info("当前时间窗口内评论数消息的处理结果为:{}", formatStr);

                        return formatStr;
                    }
                }).toStream()
                .map((key, value) -> {
                    return new KeyValue<>(key.key().toString(), formatObj(key.key().toString(), value));
                }).to(ApCommentConstants.AP_COMMENT_STREAM_OUT_SYNC_ARTICLE);

        return stream;
    }

    /**
     * 处理流式计算后的结果格式
     *
     * @param articleId
     * @param value     :  add:1
     * @return
     */
    private String formatObj(String articleId, String value) {
        Map resMap = new HashMap();
        resMap.put("articleId", articleId);
        resMap.put("add", value.split(":")[1]);

        return JSON.toJSONString(resMap);
    }

}
